#include "ThreadChannel.h"
#include "EventLoop.h"
#include "FIFOQueue.h"
#include "PlatLogger.h"
#include "Util.h"
#include "XnccEventfd.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {
    static constexpr uint32_t THREAD_MSG_HEAD_LEN = sizeof(msg::MsgHeaderInThread);
    // 线程间通信的队列和ThreadChannel唯一对应绑定
    ThreadChannel::ThreadChannel(EventLoop* loop, std::shared_ptr<msg::FIFOQueue> msgQueue)
        : Channel(loop,
                  loop->mode() != types::IO_MULTI_MODE::value::NO_IO_MULTI ? msgQueue->notifyId()->Fd() :
                                                                             msgQueue->id(),
                  types::FROM_OTHER_THREAD),
          msgQueue_(std::move(msgQueue)),
          notifyfd_(msgQueue_->notifyId())
    {
        TRACE_INFO << "create ThreadChannel bind queueId:"_s << msgQueue_->id() << " need notify:"_s
                   << (msgQueue_->notifyId() != nullptr) << " bind loop id:" << loop->id()
                   << " loop type:" << types::IO_MULTI_MODE::toString(loop->mode());
    }
    void ThreadChannel::handleRead()
    {
        readfd();  // read fd
        HandleMessages();
    }
    void ThreadChannel::HandleMessages(bool untilEmpty)
    {
        if (untilEmpty) {
            while (true) {
                bool ret = HandleSingleMessage();
                if (!ret) {
                    break;
                }
            }
        }
        else {
            HandleSingleMessage();
        }
    }
    bool ThreadChannel::readfd()
    {
        uint64_t one = 1;
        return notifyfd_->read(one) == sizeof one;
    }

    types::queue_id_t ThreadChannel::queueId() const { return msgQueue_->id(); }

    bool ThreadChannel::HandleSingleMessage()
    {
        char* readBuf = msgQueue_->getReadBuffer(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf == nullptr)) {
            if (loopType() != types::IO_MULTI_MODE::value::NO_IO_MULTI) {
                TRACE_ERROR << "thread id:" << loopId() << " receive message: but cannot get data from qeueue "
                            << msgQueue_->id();
            }
            return false;
        }
        auto* head    = reinterpret_cast<msg::MsgHeaderInThread*>(readBuf);
        char* dataBuf = msgQueue_->getReadBuffer(static_cast<types::queue_size_t>(head->len + THREAD_MSG_HEAD_LEN));
        if (likely(dataBuf)) {
            const auto& funcs = UserFuncs();
            auto        iter  = funcs.find(head->id);
            if (iter != funcs.end()) {
                iter->second(dataBuf + THREAD_MSG_HEAD_LEN, head->len, head->tid, head->connId);
            }
            else {
                TRACE_ERROR << "thread id:" << loopId() << " receive message:" << head->id
                            << " but has not registed handle func";
            }
            TRACE_DEBUG << "thread id:" << loopId() << " receive message:" << head->id
                        << " from queue id:" << msgQueue_->id();
            msgQueue_->finishRead();
            return true;
        }
        TRACE_ERROR << "thread id:" << loopId() << " receive message: but cannot get total data from qeueue "
                    << msgQueue_->id();

        return false;
    }
}  // namespace foundation
}  // namespace xncc